package com.cml.getway.core.netty.processor;

import com.cml.getway.common.enums.ResponseCode;
import com.cml.getway.core.GetwayConfig;
import com.cml.getway.core.context.HttpRequestWrapper;
import com.cml.getway.core.helper.ResponseHelper;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.FullHttpResponse;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 姓名：陈茂林
 * 时间：2023/12/20
 * 描述：
 */
public class NettyMpmcProcessor implements NettyProcessor{
    private GetwayConfig getwayConfig;
    private  NettyCoreProcessor nettyCoreProcessor;

    private ExecutorService executorService ;

    private BlockingQueue<HttpRequestWrapper> blockingQueue = new LinkedBlockingQueue();

    private boolean userPool = true;

    private volatile  boolean isRun;

    private Thread processorThead;
    public NettyMpmcProcessor(GetwayConfig getwayConfig,NettyCoreProcessor nettyCoreProcessor,boolean userPool){
        this.getwayConfig=getwayConfig;
        this.nettyCoreProcessor=nettyCoreProcessor;
        this.userPool = userPool;
    }
    @Override
    public void process(HttpRequestWrapper httpRequestWrapper) throws Exception {
        blockingQueue.put(httpRequestWrapper);
    }

    @Override
    public void start() {
        isRun=true;
        nettyCoreProcessor.start();
        if(this.userPool){
            executorService =Executors.newFixedThreadPool(getwayConfig.getProcessThread());
            for(int i=0;i<getwayConfig.getProcessThread();i++){
                executorService.submit(new ConsumerProcessorThread());
            }
        }else{
            processorThead=new ConsumerProcessorThread();
            this.processorThead.start();
        }
    }

    @Override
    public void shutdown() {
        isRun=false;
        nettyCoreProcessor.shutdown();
    }

    public   class ConsumerProcessorThread extends Thread{
        @Override
        public void run() {
            while (isRun){
                ChannelHandlerContext ctx = null;
                try {
                    HttpRequestWrapper httpRequestWrapper =blockingQueue.take();
                    ctx=httpRequestWrapper.getCtx();
                    nettyCoreProcessor.process(httpRequestWrapper);
                } catch (InterruptedException e) {
                    if(ctx!=null) {
                        FullHttpResponse fullHttpResponse = ResponseHelper.getHttpResponse(ResponseCode.INTERNAL_ERROR);
                        ctx.writeAndFlush(fullHttpResponse).addListener(ChannelFutureListener.CLOSE);
                    }
                }
            }
        }
    }

}
